package com.apps.sdses.flink141.demo;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/12/12 15:16
 */
public class SplitDemo {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //Source
        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    data_id       STRING,\n" +     //数据id
                "    stn_id        STRING,\n" +     //站点id
                "    dt   TIMESTAMP,\n" +           //数据采集时间
                "    jz1_ssll   DECIMAL(10, 2),\n" +        //机组1瞬时流量
                "    jz2_ssll   DECIMAL(10, 2),\n" +        //机组2瞬时流量
                "    jz1_ljll   DECIMAL(10, 2),\n" +        //机组1累计流量
                "    jz2_ljll   DECIMAL(10, 2)\n" +         //机组2累计流量
                ") WITH (\n" +
                "  'connector' = 'datagen', \n" +
//                "  'number-of-rows' = '5', \n" +   //一次生成100万条数据
                "  'rows-per-second' = '1', \n" +   //每秒生成10000条数据
                "  'fields.data_id.length' = '5', \n" +
                "  'fields.stn_id.length' = '5' \n" +
                ")");

//        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
//                "LIKE sourceTable (EXCLUDING ALL)");

        tableEnv.executeSql("CREATE TABLE sinkTable (" +
                "ts        BIGINT, \n" +
                "data_id        STRING, \n" +
                "stn_id         STRING, \n" +
                "dt             TIMESTAMP, \n" +
                "ssll           DECIMAL(10, 2), \n" +
                "ljll           DECIMAL(10, 2)) \n" +
                " WITH ('connector' = 'print')\n");
//                + "LIKE sourceTable (EXCLUDING ALL)");

        tableEnv.executeSql("INSERT INTO sinkTable " +
                "SELECT " +
                "  UNIX_TIMESTAMP(CAST(dt AS STRING)), " +
                "  data_id, " +
                "  stn_id, " +
                "  dt, " +
                "  jz1_ssll AS ssll, " +
                "  jz1_ljll AS ljll " +
                "FROM sourceTable " +
                "UNION ALL " +
                "SELECT " +
                "  UNIX_TIMESTAMP(CAST(dt AS STRING)), " +
                "  data_id, " +
                "  stn_id, " +
                "  dt, " +
                "  jz2_ssll AS ssll, " +
                "  jz2_ljll AS ljll " +
                "FROM sourceTable"
        );

    }
}
